package org.leiyang.common.redis;

import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.BitFieldSubCommands;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.*;

@Component
public class ReactiveRedisCache {
    private final ReactiveRedisTemplate<String, Object> redisTemplate;

    public ReactiveRedisCache(ReactiveRedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * 判断KEY是否存在
     * @param key key
     * @return 是否存在这个KEY
     */
    public Mono<Boolean> keyExist (final String key) {
        return redisTemplate.hasKey(key);
    }

    /**
     * 对value进行自增操作
     * @param key key
     * @param increase 增加的值,可以为负数或者正数
     * @return 更新之后的value值
     */
    public Mono<Long> increaseValue (final String key, final Long increase) {
        return redisTemplate.opsForValue().increment(key, increase);
    }

    public Mono<Boolean> setCacheObjectNX(final String key, final Object value, final Integer timeout)
    {
        return redisTemplate.opsForValue().setIfAbsent(key, value, Duration.ofSeconds(timeout));
    }

    /**
     * 缓存基本的对象，Integer、String、实体类等
     *
     * @param key 缓存的键值
     * @param value 缓存的值
     */
    public Mono<Boolean> setCacheObject(final String key, final Object value)
    {
        return redisTemplate.opsForValue().set(key, value);
    }

    /**
     * 缓存基本的对象，Integer、String、实体类等
     *
     * @param key 缓存的键值
     * @param value 缓存的值
     * @param timeout 时间
     */
    public Mono<Boolean> setCacheObject(final String key, final Object value, final Integer timeout)
    {
        return redisTemplate.opsForValue().set(key, value, Duration.ofSeconds(timeout));
    }

    /**
     * 设置有效时间
     *
     * @param key Redis键
     * @param timeout 超时时间
     * @return true=设置成功；false=设置失败
     */
    public Mono<Boolean> expire(final String key, final long timeout)
    {
        return redisTemplate.expire(key, Duration.ofSeconds(timeout));
    }

    /**
     * 如果没有过期时间，则设置过期时间
     * @param key 键
     * @param timeout 超时时间
     */
    public void expireIfNone(final String key, final long timeout) {
        Mono<Duration> expire = redisTemplate.getExpire(key);
        expire.defaultIfEmpty(Duration.ZERO).subscribe(duration -> {
            //表示没有设置过期时间
            if(Objects.equals(duration, Duration.ZERO)) {
                expire(key, timeout);
            }
        });
    }

    /**
     * 获得缓存的基本对象。
     * @param key 缓存键值
     * @return 缓存键值对应的数据
     */
    public Mono<Object> getCacheObject(final String key)
    {
        ReactiveValueOperations<String, Object> operation = redisTemplate.opsForValue();
        return operation.get(key);
    }

    /**
     * 删除单个对象
     * @param key 待删除的key
     */
    public Mono<Long> deleteObject(final String key)
    {
        return redisTemplate.delete(key);
    }

    /**
     * 删除集合对象
     * @param key 多个KEY
     * @return 成功删除的KEY个数
     */
    public Mono<Long> deleteKeys(final String... key)
    {
        return redisTemplate.delete(key);
    }

    /**
     * 将List数据追加到原来的集合后面
     * @param key 缓存的键值
     * @param dataList 待缓存的List数据
     * @return 缓存的对象
     */
    public Mono<Long> setCacheList(final String key, final List<Object> dataList)
    {
        return redisTemplate.opsForList().rightPushAll(key, dataList);
    }

    /**
     * 获得缓存的list对象
     *
     * @param key 缓存的键值
     * @return 缓存键值对应的数据
     */
    public Flux<Object> getCacheList(final String key)
    {
        return redisTemplate.opsForList().range(key, 0, -1);
    }

    /**
     * 缓存Set
     *
     * @param key 缓存键值
     * @param dataSet 缓存的数据
     * @return 成功加入的元素个数
     */
    public Mono<Long> setCacheSet(final String key, final Set<Object> dataSet)
    {
        return redisTemplate.opsForSet().add(key, dataSet.toArray());
    }

    /**
     * 获得缓存的set
     * @param key 缓存键值
     * @return values
     */
    public Flux<Object> getCacheSet(final String key)
    {
        return redisTemplate.opsForSet().members(key);
    }

    /**
     * 检查对应的值在dataset中是否存在
     * @param key 缓存键值
     * @param value 缓存的数据
     * @return 是否存在
     */
    public Mono<Boolean> isExistInSet(final String key, final Object value) {
        return redisTemplate.opsForSet().isMember(key, value);
    }

    /**
     * 批量删除set里的元素
     * @param key 缓存键值
     * @param values  待删除的元素值
     */
    public Mono<Long> deleteValuesFromCacheSet(final String key, final Object[] values) {
        return redisTemplate.opsForSet().remove(key, values);
    }

    /**
     * 缓存Map
     *
     * @param key
     * @param dataMap
     */
    public Mono<Boolean> setCacheMap(final String key, final Map<String, Object> dataMap)
    {
        if (dataMap != null) {
            return redisTemplate.opsForHash().putAll(key, dataMap);
        }
        return Mono.just(Boolean.TRUE);
    }

    /**
     * 缓存Map
     *  V1版本
     * @param key
     * @param dataMap
     */
    public Mono<Boolean> setCacheMapV1(final String key, final Map<Object, Object> dataMap)
    {
        if (dataMap != null) {
            return redisTemplate.opsForHash().putAll(key, dataMap);
        }
        return Mono.just(Boolean.TRUE);
    }

    /**
     * 获得缓存的Map
     * @param key KEY
     * @return map
     */
    public Flux<Map.Entry<Object, Object>> getCacheMap(final String key)
    {
        return redisTemplate.opsForHash().entries(key);
    }

    /**
     * 往Hash中存入数据
     *
     * @param key Redis键
     * @param hKey Hash键
     * @param value 值
     */
    public Mono<Boolean> setCacheMapValue(final String key, final String hKey, final Object value)
    {
        return redisTemplate.opsForHash().put(key, hKey, value);
    }


    /**
     * 对hash值进行自增操作
     * @param key Redis键
     * @param hKey Hash键
     * @param incrementValue 自增的值
     * @return updated value
     */
    public Mono<Long> incrementCacheMapValue(final String key, final String hKey, final Long incrementValue)
    {
        return redisTemplate.opsForHash().increment(key, hKey, incrementValue);
    }

    /**
     * 获取Hash中的数据
     * @param key Redis键
     * @param hKey Hash键
     * @return Hash中的对象
     */
    public Mono<Object> getCacheMapValue(final String key, final String hKey)
    {
        return redisTemplate.opsForHash().get(key, hKey);
    }

    /**
     * 删除Hash中的数据
     * @param key KEY
     * @param hkey hash的键
     */
    public Mono<Long> delCacheMapValue(final String key, final String hkey)
    {
        return redisTemplate.opsForHash().remove(key, hkey);
    }

    /**
     * 获取多个Hash中的数据
     *
     * @param key Redis键
     * @param hKeys Hash键集合
     * @return Hash对象集合
     */
    public Mono<List<Object>> getMultiCacheMapValue(final String key, final Collection<Object> hKeys)
    {
        return redisTemplate.opsForHash().multiGet(key, hKeys);
    }

    /**
     * 获得缓存的基本对象列表
     * @param pattern 字符串前缀
     * @return 对象列表
     */
    public Flux<String> keys(final String pattern)
    {
        return redisTemplate.keys(pattern);
    }

    /**
     * scan方式获取指定前缀的key列表
     * @param pattern 带*的前缀
     * @param count 期望本次scan的最大key的数量
     * @return key列表
     */
    public Flux<String> scan(final String pattern, final Long count) {
        ScanOptions scanOptions = ScanOptions.scanOptions().match(pattern).count(count).build();
        return redisTemplate.scan(scanOptions);
    }

    /**
     * 向Zset中添加元素
     * @param key zset的key
     * @param memberName 元素名
     * @param score 分数
     * @return 是否添加成功
     */
    public Mono<Boolean> addToZset(final String key, final String memberName, final Double score) {
        return redisTemplate.opsForZSet().add(key, memberName, score);
    }

    /**
     * 批量添加zset元素
     * @param key zset的key
     * @param tuples 批量元素
     * @return 返回添加成功的条数
     */
    public Mono<Long> addBatchToZset(final String key, final Set<ZSetOperations.TypedTuple<Object>> tuples) {
        return redisTemplate.opsForZSet().addAll(key, tuples);
    }

    /**
     * 从Zset中删除指定的元素
     * @param key zset的key
     * @param memberName 元素名
     * @return 返回删除的条数
     */
    public Mono<Long> deleteOneFromZset(final String key, final String memberName) {
        return redisTemplate.opsForZSet().remove(key, memberName);
    }

    /**
     * 返回集合内元素的排名，以及分数（从大到小）
     * @param key zset的key
     * @param startIndex 起始索引值
     * @param endIndex 尾部索引值
     * @return 元素列表
     */
    public Flux<ZSetOperations.TypedTuple<Object>> listAllMembersFromZset(final String key, final long startIndex, final long endIndex) {
        return redisTemplate.opsForZSet().reverseRangeWithScores(key, Range.closed(startIndex, endIndex));
    }

    /**
     * 返回集合内的成员个数
     * @param key zset的key
     * @return 个数
     */
    public Mono<Long> countOfZset (final String key) {
        return redisTemplate.opsForZSet().size(key);
    }

    /**
     * 获得指定元素的分数
     * @param key zset的key
     * @param memberName 元素名
     * @return 指定元素的分数值
     */
    public Mono<Double> getScoreOfOneMember(final String key, final String memberName) {
        return redisTemplate.opsForZSet().score(key, memberName);
    }

    /**
     * 给指定的元素打分
     * @param key zset的key
     * @param memberName 元素名
     * @param newScore 新的分数
     */
    public void setScoreOfOneMember(final String key, final String memberName, final Double newScore) {
        //注：存在并发问题，可以使用lua方式保证其原子性
        getScoreOfOneMember(key, memberName).subscribe(score -> {
            //获取本次分数的增量值-offset，可以为负数
            redisTemplate.opsForZSet().incrementScore(key, memberName, newScore - score).subscribe();
        });
    }

    /**
     * 删除指定索引范围的元素
     * @param key zset的key
     * @param startIndex 起始索引值
     * @param endIndex 尾部索引值
     * @return 删除成功的元素个数
     */
    public Mono<Long> deleteByIndexs(final String key, final long startIndex, final long endIndex) {
        return redisTemplate.opsForZSet().removeRange(key, Range.closed(startIndex, endIndex));
    }

    /**
     * 设置bitmap的值
     * @param key 键
     * @param offset  偏移量
     * @param result  true or false
     * @return  是否设置成功
     */
    public Mono<Boolean> setBitValue(final String key, final long offset, final boolean result) {
        return redisTemplate.opsForValue().setBit(key, offset, result);
    }

    /**
     * 获取bitmap里对应key的offset位置上的值
     * @param key 键
     * @param offset 偏移量
     * @return  获取到的值 true or false
     */
    public Mono<Boolean> getBitValue(final String key, final long offset) {
        return redisTemplate.opsForValue().getBit(key, offset);
    }

    /**
     * 返回指定的二进制范围
     * @param key 键
     * @param limit 最大长度
     * @param offset 偏移量，开始位置
     * @return 二进制的列表
     */
    public Mono<List<Long>> getBitFields(final String key, final int limit, final long offset) {
        return redisTemplate.opsForValue().bitField(key, BitFieldSubCommands.create().get(BitFieldSubCommands.BitFieldType
                .unsigned(limit)).valueAt(offset));
    }

    /**
     * 批量设置bitmap的多个点位
     * @param key bitmap的KEY
     * @param offsets 多个点位值
     * @param result true or false
     * @return 格式: offset : boolean
     */
    public Mono<Map<Long, Boolean>> batchSetBitValues(final String key, final List<Long> offsets, final boolean result) {
        Flux<Boolean> results = redisTemplate.executeInSession(operations ->
                Flux.fromIterable(offsets)
                        .flatMap(offset -> operations.opsForValue().setBit(key, offset, result))
        );

        return results.collectList().map(data -> {
            if(org.springframework.util.CollectionUtils.isEmpty(data)) {
                return Collections.emptyMap();
            }
            Map<Long, Boolean> map = new HashMap<>(offsets.size());
            for (int i = 0; i < offsets.size(); i++) {
                map.put(offsets.get(i), data.get(i));
            }
            return map;
        });
    }

    /**
     * 批量获取bitmap的多个点位值
     * @param key bitmap的KEY
     * @param offsets 多个点位值
     * @return 格式: offset : boolean
     */
    public Map<Long, Boolean> batchGetBitValues(final String key, final List<Long> offsets) {
        Flux<Boolean> results = redisTemplate.executeInSession(operations ->
                Flux.fromIterable(offsets)
                        .flatMap(offset -> operations.opsForValue().getBit(key, offset))
            );

        List<Boolean> data = results.collectList().block();
        if(org.springframework.util.CollectionUtils.isEmpty(data)) {
            return Collections.emptyMap();
        }
        Map<Long, Boolean> map = new HashMap<>(offsets.size());
        for (int i = 0; i < offsets.size(); i++) {
            map.put(offsets.get(i), data.get(i));
        }
        return map;
    }

    /**
     * 向HyperLogLog中新增一个元素
     * @param key 键
     * @param member 元素
     * @return 新增成功的元素个数
     */
    public Mono<Long> hyperLogLogAdd(final String key, final Object member) {
        ReactiveHyperLogLogOperations<String, Object> vo = redisTemplate.opsForHyperLogLog();
        return vo.add(key, member);
    }

    /**
     * 获取指定的key的基数的和
     * @param keys key列表
     * @return 基数和
     */
    public Mono<Long> countOneHyperLogLog(final String... keys) {
        return redisTemplate.opsForHyperLogLog().size(keys);
    }

    /**
     * 生成redis版本的MQ的操作类
     */
    public ReactiveStreamOperations<String, String, String> createStreamOperations() {
        return redisTemplate.opsForStream();
    }

    /**
     * 执行lua脚本
     * @param luaScript lua脚本内容
     * @param keys key入参
     * @return 执行的
     * @param <T> LUA脚本的返回类型
     */
    public <T> Flux<T> executeLua(String luaScript, List<String> keys, Class<T> resultType) {
        DefaultRedisScript<T> script = new DefaultRedisScript<>();
        script.setScriptText(luaScript);
        script.setResultType(resultType);
        return redisTemplate.execute(script, keys);
    }
}